#!/usr/bin/env python
# -*- coding: UTF-8 -*-
"""
@Project ：python常用模块库 
@File ：file_monitor.py
@IDE  ：PyCharm 
@Author ：李涵彬
@Date ：2025/2/21 上午9:31 
"""

"""
Industrial-Grade File System Monitor
Version: 1.0
Features:
- 文件数量监控
- 目录大小监控
- 自动清理旧文件
- 安全删除机制
- 实时日志记录
"""

import os
import time
import logging
import platform
from pathlib import Path
from datetime import datetime
from typing import List, Tuple, Dict


class FileMonitor:
	"""文件系统监控与自动清理器"""

	def __init__(self,
				 target_dir: str,
				 max_files: int = 1000,
				 max_size_gb: float = 10.0,
				 check_interval: int = 300,
				 exclude_patterns: List[str] = None):
		"""
		初始化监控器
		:param target_dir: 监控目录路径
		:param max_files: 最大文件数量限制
		:param max_size_gb: 最大目录大小(GB)
		:param check_interval: 检查间隔(秒)
		:param exclude_patterns: 排除文件模式列表
		"""
		self.monitor_path = Path(target_dir).resolve()
		self.max_files = max_files
		self.max_size = int(max_size_gb * 1024 ** 3)  # 转换为字节
		self.check_interval = check_interval
		self.exclude_patterns = exclude_patterns or []
		self._setup_logging()
		self._validate_directory()

	def _setup_logging(self):
		"""配置日志系统"""
		logging.basicConfig(
			level=logging.INFO,
			format='%(asctime)s [%(levelname)s] %(message)s',
			datefmt='%Y-%m-%d %H:%M:%S',
			handlers=[
				logging.FileHandler('file_monitor.log'),
				logging.StreamHandler()
			]
		)
		self.logger = logging.getLogger(__name__)

	def _validate_directory(self):
		"""验证目标目录有效性"""
		if not self.monitor_path.exists():
			raise FileNotFoundError(f"目标目录不存在: {self.monitor_path}")
		if not self.monitor_path.is_dir():
			raise NotADirectoryError(f"路径不是目录: {self.monitor_path}")
		if platform.system() == 'Linux' and os.stat(self.monitor_path).st_dev != os.stat('/').st_dev:
			self.logger.warning("监控目录位于独立文件系统，请注意存储限制")

	def start_monitoring(self):
		"""启动监控循环"""
		self.logger.info(f"启动目录监控: {self.monitor_path}")
		try:
			while True:
				self._check_and_clean()
				time.sleep(self.check_interval)
		except KeyboardInterrupt:
			self.logger.info("监控服务已手动停止")
		except Exception as e:
			self.logger.critical(f"监控服务异常终止: {str(e)}", exc_info=True)

	def _check_and_clean(self):
		"""执行检查并清理"""
		file_list = self._scan_directory()
		current_count = len(file_list)
		current_size = sum(f['size'] for f in file_list)

		# 记录当前状态
		self.logger.info(f"当前状态: 文件数 {current_count}/{self.max_files} | "
						 f"目录大小 {current_size / 1024 ** 3:.2f}GB/{self.max_size / 1024 ** 3:.2f}GB")

		# 检查是否需要清理
		if current_count <= self.max_files and current_size <= self.max_size:
			return

		self._perform_cleanup(file_list)

	def _scan_directory(self) -> List[Dict]:
		"""扫描目录并获取文件列表"""
		file_list = []
		for entry in self.monitor_path.rglob('*'):
			if self._should_skip(entry):
				continue

			try:
				stat = entry.stat()
				file_list.append({
					'path': entry,
					'size': stat.st_size,
					'ctime': stat.st_ctime,
					'is_dir': entry.is_dir()
				})
			except Exception as e:
				self.logger.error(f"无法获取文件信息: {entry} - {str(e)}")

		# 按创建时间排序（旧文件在前）
		return sorted(file_list, key=lambda x: x['ctime'])

	def _should_skip(self, path: Path) -> bool:
		"""判断是否应跳过该路径"""
		if not path.exists():
			return True
		if any(path.match(p) for p in self.exclude_patterns):
			return True
		if path.is_symlink():
			return True
		return False

	def _perform_cleanup(self, file_list: List[Dict]):
		"""执行清理操作"""
		deleted_count = 0
		freed_space = 0
		current_count = len(file_list)
		current_size = sum(f['size'] for f in file_list)

		for file_info in file_list:
			if current_count <= self.max_files and current_size <= self.max_size:
				break

			try:
				self._safe_delete(file_info)
				deleted_count += 1
				freed_space += file_info['size']
				current_count -= 1
				current_size -= file_info['size']
			except Exception as e:
				self.logger.error(f"删除失败: {file_info['path']} - {str(e)}")

		self.logger.warning(
			f"清理完成: 删除项目 {deleted_count} 个, "
			f"释放空间 {freed_space / 1024 ** 3:.2f}GB"
		)

	def _safe_delete(self, file_info: Dict):
		"""安全删除文件/目录"""
		target = file_info['path']
		if file_info['is_dir']:
			self._delete_directory(target)
		else:
			target.unlink()
		self.logger.info(f"已删除: {target} (创建时间: {self._format_time(file_info['ctime'])})")

	def _delete_directory(self, dir_path: Path):
		"""递归删除目录"""
		for item in dir_path.iterdir():
			if item.is_dir():
				self._delete_directory(item)
			else:
				item.unlink()
		dir_path.rmdir()

	@staticmethod
	def _format_time(timestamp: float) -> str:
		"""格式化时间戳"""
		return datetime.fromtimestamp(timestamp).strftime('%Y-%m-%d %H:%M:%S')


if __name__ == "__main__":
	# #################### 配置区域 ####################
	MONITOR_CONFIG = {
		'target_dir': './test_dir',  # 监控目录
		'max_files': 100,  # 最大文件数
		'max_size_gb': 1.0,  # 最大目录大小(GB)
		'check_interval': 300,  # 检查间隔(秒)
		'exclude_patterns': ['*.tmp', '*.lock']  # 排除文件模式
	}
	# #################################################

	try:
		monitor = FileMonitor(**MONITOR_CONFIG)
		monitor.start_monitoring()
	except Exception as e:
		logging.error(f"监控服务启动失败: {str(e)}", exc_info=True)